package com.shujia.flink.state;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Demo1State {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> wordsDS = env.socketTextStream("master", 8888);

        //分组
        KeyedStream<String, String> keyByDS = wordsDS.keyBy(word -> word);

        /*
         * process算子时flink提供的一个底层算子，可以获取到flink底层的状态，时间和数据
         */
        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .process(new KeyedProcessFunction<String, String, Tuple2<String, Integer>>() {
                    //保存之前统计的结果（状态）
                    //问题：同一个task中的数据共享同一个count变量
                    //int count = 0;
                    //需要为每一个key保存一个结果
                    //使用单词作为key,数量作为value
                    //问题：使用hashmap保存计算的中间结果，flink的checkpoint不会将hashmap中的数据持久化到hdfs总
                    //所以任务失败重启会丢失之前的结果
                    final HashMap<String, Integer> map = new HashMap<>();
                    /**
                     * processElement方法每一条数据执行一次
                     * @param word 一行数据
                     * @param ctx 上下文对象,可以获取到flink的key和时间属性
                     * @param out 用于将处理结果发送到下游
                     */
                    @Override
                    public void processElement(String word,
                                               KeyedProcessFunction<String, String, Tuple2<String, Integer>>.Context ctx,
                                               Collector<Tuple2<String, Integer>> out) throws Exception {

                        System.out.println(map);
                        //1、通过key获取value
                        //获取之前的结果（状态）
                        Integer count = map.getOrDefault(word, 0);
                        //基于之前的结果进行计算
                        count++;
                        //将计算结果发送到下游
                        out.collect(Tuple2.of(word, count));
                        //更新之前的结果
                        map.put(word, count);
                    }
                });

        countDS.print();

        env.execute();
    }
}
